package com.mjf.spark.day03

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换算子-mapPartitions
 */
object Spark01_Transformation_mapPartitions {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark01_Transformation_mapPartitions")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)

    // 以分区为单位，对RDD中的数据进行映射
    // 一般适用于批处理的操作，比如：将RDD中的元素插入到数据库中，需要数据库连接
    // 如果每个元素都创建一个连接，效率很低。可以对每个分区的元素创建一个连接
    val newRDD: RDD[Int] = rdd.mapPartitions(datas => {
      datas.map(_ * 2)    // 此处map不是算子，是迭代器的方法，算子是rdd调用的方法
    })

    newRDD.collect().foreach(println)

    // 关闭连接
    sc. stop()

  }
}
